#include "Server.h"
#include "Log.h"

#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <unistd.h>
#include <fcntl.h>

#include <functional>

using namespace std;

Server::Server(const string &name, const string &ip, const int port)
:_listenThread(bind(&Server::runInThread,this), name), _threadPool(name), _started(false), _listenAddr(ip, port), _name(name){
	pthread_mutex_init(&_connectMapMutex, NULL);
	pthread_cond_init(&_connectMapCond, NULL);
}

Server::~Server(){
	if(_started){
		_started = false;
		_listenThread.join();
	}
//	_threadPool.stop();
	close(_listenSock);
	close(_epfd);
	pthread_mutex_destroy(&_connectMapMutex);
	pthread_cond_destroy(&_connectMapCond);
}

int Server::init(){
	_epfd = epoll_create(1024);
	LOG_DEBUG_VA("epoll fd create epfd: %d", _epfd);
	if (-1 == _epfd) {
		LOG_ERROR_VA("epoll fd create error.");
		return -1;
	}
	int ret = 0;
	_listenSock = socket(AF_INET, SOCK_STREAM, 0);
	LOG_DEBUG_VA("listen socket: %d", _listenSock);
	if (_listenSock < 0) {
		LOG_ERROR_VA("listening socket open error.");
		return -1;
	}
	int optval = 1;
	ret = setsockopt(_listenSock, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(int));
	if (ret < 0) {
		LOG_ERROR_VA("set socket option error.");
		return -1;
	}
	ret = ::bind(_listenSock, (struct sockaddr *)(&_listenAddr.getAddr()), sizeof(_listenAddr.getAddr()));
	if (ret < 0) {
		LOG_ERROR_VA("bind error.");
		Socket::tcpClose(_listenSock);
		return -1;
	}
	int sockFlag = fcntl(_listenSock, F_GETFL, 0);
	fcntl(_listenSock, F_SETFL, sockFlag | O_NONBLOCK);
	ret = listen(_listenSock, 128);
	if (ret < 0) {
		char errbuf[1024];
		strerror_r(errno, errbuf, 1023);
		LOG_ERROR_VA("listening socket error: %s", errbuf);
		Socket::tcpClose(_listenSock);
		return -1;
	}
	_eplistenevent.data.u64 = 0UL;
	_eplistenevent.data.fd = _listenSock;
	//_eplistenevent.events = EPOLLIN | EPOLLRDHUP;
	_eplistenevent.events = EPOLLIN;
	ret = epoll_ctl(_epfd, EPOLL_CTL_ADD, _listenSock, &_eplistenevent);
	if (ret < 0) {
		LOG_ERROR_VA("epoll add listen sock error.");
		Socket::tcpClose(_listenSock);
		return -1;
	}
	
	return ret;
}

//thread entrance for this server.
int Server::runInThread(){
	int ret = 0;
	LOG_DEBUG_VA("server %s listen thread start", _name.c_str());
	while (_started) {
		ret = checkTask();
		if (ret < 0) {
			LOG_ERROR_VA("check task error for %s", _name.c_str());
		}
	}
	return 0;
}

int Server::checkTask(){
	int ret = 0;
	LOG_DEBUG_VA("server %s listen thread TcpEpollWait...", _name.c_str());
	int numfd = Socket::tcpEpollWait(_epfd, _epevents, 1024, -1);
	if (numfd < 0) {
		if(errno == EINTR) {
			return 0;
		}else{
			LOG_ERROR_VA("Server::checkTask() tcpEpollWait failed.");
			return numfd;
		}
	}
	//dispatch ready fd
	struct sockaddr_in peerAddr;
	socklen_t peerAddrLen = sizeof(peerAddr);
	LOG_DEBUG_VA("server %s epoll_wait num: %d", _name.c_str(), numfd);
	for (int i = 0; i < numfd; ++i) {
		if (_epevents[i].events & EPOLLIN) {
			if(_epevents[i].data.fd == _listenSock) {
				std::vector<Address> peerAddrVec;
				std::vector<int> peerFdVec;
				int ret = Socket::tcpAccept(_listenSock, peerFdVec, peerAddrVec);
				if (ret < 0) {
					LOG_ERROR_VA("one task fd failed to accept.");
					continue;
				}
				TcpConnect *tc = NULL;
				for(int i = 0; i < peerFdVec.size(); i++){
					tc = prepareConnect(peerFdVec[i], peerAddrVec[i]);
					if (NULL == tc) {
						LOG_ERROR_VA("server %s refuse to accept fd: %d ", _name.c_str(), peerFdVec[i]);
						Socket::tcpClose(peerFdVec[i]);
						continue;
					}
				/*
					int sockFlag = fcntl(peerFdVec[i], F_GETFL, 0);
					fcntl(peerFdVec[i], F_SETFL, sockFlag | O_NONBLOCK);
					struct epoll_event ev;
					ev.data.ptr = &peerAddrVec[i];
					ev.data.u64 = 0UL;
					ev.data.fd = peerFdVec[i];
					ev.events = EPOLLIN;
					int ret = Socket::tcpEpollAdd(_epfd, peerFdVec[i], &ev);
					if (-1 == ret) {
						LOG_ERROR_VA("server %s add epoll failed fd: %d", _name.c_str(), peerFdVec[i]);
						Socket::tcpClose(peerFdVec[i]);
						continue;
					}else{
						ret = getThreadPool().addTcpConnect(tc);
						LOG_DEBUG_VA("server %s accepted TcpConnect %x with <%s:%d>: %d", _name.c_str(), tc, peerAddrVec[i].getIP().c_str(), peerAddrVec[i].getPort(), peerFdVec[i]);
					}
				*/
					ret = getThreadPool().addTcpConnect(tc);
					if(0 == ret){
						LOG_DEBUG_VA("server %s accepted TcpConnect %x with <%s:%d>: %d", _name.c_str(), tc, peerAddrVec[i].getIP().c_str(), peerAddrVec[i].getPort(), peerFdVec[i]);
					}else{
						LOG_ERROR_VA("server %s getThreadPool().addTcpConnect() %x failed ", _name.c_str(), tc);
						resetConnect(peerFdVec[i]);
						Socket::tcpClose(peerFdVec[i]);
						continue;
					}
				}
			}else{
				LOG_DEBUG_VA("sever %s can receive from fd %d", _name.c_str(), _epevents[i].data.fd);
			}
		}else if(_epevents[i].events & EPOLLHUP) {
			LOG_ERROR_VA("fd: %d get EPOLLHUP.", _epevents[i].data.fd);
			resetConnect(_epevents[i].data.fd);
		} else if (_epevents[i].events & EPOLLERR) {
			LOG_ERROR_VA("fd: %d get EPOLLERR.", _epevents[i].data.fd);
			resetConnect(_epevents[i].data.fd);
		} else if (_epevents[i].events & EPOLLOUT) {
			LOG_ERROR_VA("fd: %d get EPOLLOUT.", _epevents[i].data.fd);
			//doWriteEvent(_epevents[i].data.fd);
		} else {
			LOG_ERROR_VA("fd: %d get something wrong.", _epevents[i].data.fd);
			resetConnect(_epevents[i].data.fd);
		}
	}
	return ret;
}

int Server::run(){
	LOG_INFO_VA("server starting...");
	int ret;
	
	ret = init();
	if(ret != 0)
		return -1;
	_started = true;
	_listenThread.start();

	_threadPool.setServer(this);
	_threadPool.start(20);
	return 0;
}

TcpConnect* Server::prepareConnect(int sockfd, Address &peerAddr){
	auto iter = _connectMap.find(sockfd);
	if(iter != _connectMap.end()){
		//socket fd is in use, can't accept
		//how about thr Socket Status???
		LOG_ERROR_VA("server %s can't add TcpConnect with fd %d ", _name.c_str(), sockfd);
		return NULL;
	}else{
		TcpConnect *c = new TcpConnect(sockfd, peerAddr);
		c->setConnectStatus(READY);
		struct timeval t;
		gettimeofday(&t, NULL);
		c->setConnectTime(t);
		c->setLastActiveTime(t);
		std::pair<map<int, TcpConnect*>::iterator, bool> insertPair;
//		mutexLock();
		insertPair = _connectMap.insert(std::map<int, TcpConnect *>::value_type(sockfd, c));
//		mutexUnLock();
		if(insertPair.second != true){
			delete c;
			return NULL;
		}
		LOG_DEBUG_VA("server %s add TcpConnect %x with fd %d ", _name.c_str(), c, sockfd);
		return c;
	}
}

TcpConnect *Server::getConnect(int sockfd){
	auto iter = _connectMap.find(sockfd);
	if(iter == _connectMap.end()){
		//socket fd not found
		return NULL;
	}else{
		return iter->second;
	}
}

int Server::resetConnect(int sockfd){
	//std::map<int, TcpConnect *>::iterator iter;
	auto iter = _connectMap.find(sockfd);
	if(iter == _connectMap.end()){
		//socket fd not found, can't reset
		return -1;
	}
	iter->second->close();
	//pthread_mutex_lock(&_sockMapMutex);
	mutexLock();
	_connectMap.erase(iter);
	//pthread_mutex_unlock(&_sockMapMutex);
	mutexUnLock();
	return 0;
}


